package org.zjt.rxjava.lession.first;


import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.functions.Consumer;
import io.reactivex.observables.ConnectableObservable;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static io.reactivex.internal.operators.flowable.FlowableBlockingSubscribe.subscribe;

/**
 * 1、do.... : 在执行某个方法的时候执行。但不能触发执行某个方法。
 *
 * @author zhangjuntao
 */
@Slf4j
public class SchedulerTest {

    /**
     * defer() 推迟 ：延迟创建Observable。
     * <p>
     * 当Observable被subscribe订阅后，才会执行defer中的代码。
     * <p>
     * PS:效率不搞：
     * 唯一的缺点defer()就是Observable每当你收到订阅者时都会创建一个新的。create()可以为每个用户使用相同的功能，所以它更有效率。与往常一样，衡量性能并在必要时进行优化。
     */
    @Test
    public void testDefer() {
        AtomicInteger integer = new AtomicInteger(10);
        Observable<Integer> observable = Observable.defer(() -> Observable.just(integer.get()));
        integer.set(20);
        observable.subscribe(System.out::println);
    }

    /**
     * just() : 实时创建一个Observable。
     */
    @Test
    public void testJust() {
        AtomicInteger integer = new AtomicInteger(10);
        Observable<Integer> observable = Observable.just(integer.get());
        integer.set(20);
        observable.subscribe(System.out::println);
    }


    /**
     * create() : 实时创建一个Observable。 可复用
     * <p>
     * <p>
     * PS: 效率高
     * <p>
     * create()可以为每个用户使用相同的功能，所以它更有效率。
     */
    @Test
    public void testCreate() {
        AtomicInteger integer = new AtomicInteger(10);
        Observable<Integer> observable = Observable.create(observableEmitter -> {
            observableEmitter.onNext(integer.get());
            observableEmitter.onError(new Exception("--------2332---------"));
            observableEmitter.onComplete();

        });
        integer.set(20);
        observable
                .doOnError(System.out::println)
                .onErrorReturn((throwable) -> new Integer(404))
                .doFinally(() -> System.out.println("----------Finally--------"))
                .doOnSubscribe((Consumer) o -> System.out.println("----------doOnSubscribe--------"))
                .doOnTerminate(() -> System.out.println("----------observable的生命周期结束：doOnTerminate--------"))
                .doOnNext((a) -> System.out.println("----------doOnNext--------" + a))
                .doOnComplete(() -> System.out.println("----------doOnComplete--------"))
                .doOnEach((event) -> System.out.println("event:" + event))

                .subscribe(System.out::println);   // 执行的action代码
    }


    public Observable<Integer> processor(final Integer formula) {

        return Observable.create(observableEmitter -> {
            observableEmitter.onNext(formula + 1 + 1 / 0);
            observableEmitter.onComplete();

        });
    }


    public Observable<Integer> processor2(final Integer formula) {

        return Observable.just(formula + 1 + 1 / 0);
    }


    /**
     * Observable.create 才能被拦截异常返回。 just不能拦截返回值。
     * <p>
     * retry 得到未捕获异常重试。
     */
    @Test
    public void testException() {
        processor(100)
                .doOnError(System.err::println)
                .retry(3)                           //当检测到未捕获的异常时候，直接retry。
                .onErrorReturn((e) -> -1)
                .doFinally(() -> System.out.println("----------Finally--------"))
                .subscribe(System.out::println);


        processor(100)
                .onErrorReturn((e) -> -1)
                .subscribe(System.out::println);


        processor2(100)
                .onErrorReturn((e) -> -1)
                .subscribe(System.out::println);

    }


    public Observable<List<Integer>> fromArray() {

        return Observable.fromArray(Arrays.asList(1, 2, 4, 2, 1, 2, 5));
    }


    @Test
    public void testFromArray() {
        fromArray()
                .doOnError(System.err::println)
                .retry(3)
                .buffer(2)
/*
                .scan((sum,item)->{sum+item})
*/
                .flatMap((array) -> Observable.just(array.stream().flatMap(item -> item.stream()).collect(Collectors.toList())))
/*
                .window(1, TimeUnit.SECONDS)
*/
                .subscribe(System.out::println);
    }

    /**
     * jdk flatMap 拆分数组。
     */
    @Test
    public void arrayFlatMap() {
        List<Integer> list = new ArrayList<Integer>();
        for (int i = 0; i < 10; i++) {
            list.add(new Random().nextInt(10));
        }
        List<List<Integer>> data = new ArrayList<>(1);
        final List<Integer> collect = data.stream().flatMap(integers -> integers.stream()).collect(Collectors.toList());
        collect.forEach(System.out::println);
    }


    /**
     * debounce 闲置时候发送:
     *  仅在过了一段指定的时间还没发射数据时才发射一个数据
     */
    @Test
    public void arrayDebounce() {
        Observable.create((emitter)->{
            emitter.onNext(123);
            Thread.sleep(2000L);
            emitter.onComplete();
        }).debounce(1,TimeUnit.SECONDS).subscribe(System.out::println);

        ConnectableObservable<String> source = Observable.just("Alpha","Beta","Gamma","Delta","Epsilon").publish();

        //Set up observer 1
        source.subscribe(s -> System.out.println("Observer 1: " + s));
        //Set up observer 2
        source.map(String::length).subscribe(i -> System.out.println("Observer 2: " + i));
        //Fire!
        source.connect();
    }


    /**
     * 发送器： 多个订阅器模式
     *
     * 主线程 Thread[main,5,main]
     */
    @Test
    public void connectPublish() {

        Observable<String> just = Observable.just("Alpha", "Beta", "Gamma", "Delta", "Epsilon");
        ConnectableObservable<String> source = just.publish();

        //Set up observer 1
        source.subscribe(s -> System.out.println("Observer 1: " + s +"   "+ Thread.currentThread()));

        //Set up observer 2
        source.map(String::length).subscribe(i -> System.out.println("Observer 2: " + i +"   "+ Thread.currentThread()));

        just.subscribe(i -> System.out.println("Observer 3: " + i +"   "+ Thread.currentThread()));

        //发送!
        source.connect();
    }


    /**
     * 开启新线程定时发送
     *
     * ConnectableObservable : 同一个ConnectableObservable （publish）为同一个线程
     * Observable：不同的线程。
     *
     * @throws InterruptedException
     */
    @Test
    public void intervalTest() throws InterruptedException {
        Observable<Long> interval = Observable.interval(1, TimeUnit.SECONDS);
        interval.subscribe(s -> System.out.println(s + " Mississippi - 1 - " + Thread.currentThread()));

        Thread.sleep(5000);
        interval.subscribe(s -> System.out.println(s + " Mississippi - 2 - " + Thread.currentThread()));
        log.info(Thread.currentThread().toString());


        ConnectableObservable<Long> publish = interval.publish();
        publish.subscribe(s -> System.out.println(s + " Mississippi - 3 - " + Thread.currentThread()));

        Thread.sleep(5000);
        publish.subscribe(s -> System.out.println(s + " Mississippi - 4 - " + Thread.currentThread()));

        publish.connect();

        // 发送!
        Thread.sleep(5000);
    }


    /**
     * defer 在订阅时候才会执行发布器中的代码逻辑，延迟执行
     *
     * defer 延缓
     *
     * @throws InterruptedException
     */
    @Test
    public void deferTest() throws InterruptedException {

        AtomicInteger start = new AtomicInteger(0) ,count = new AtomicInteger(5) ;

        Observable<Integer> source = Observable.range(start.get(),count.get());
        source.subscribe(i -> System.out.println("Observer 1: " + i));

        //modify count
        count.set(8);
        source.subscribe(i -> System.out.println("Observer 2: " + i));



        source = Observable.defer(() -> Observable.range(start.get(),count.get()));
        source.subscribe(i -> System.out.println("Observer 4: " + i));

        //modify count
        count.set(10);
        source.subscribe(i -> System.out.println("Observer 3: " + i));

    }


    /**
     * 来自一个callable 的返回值
     * @throws InterruptedException
     */
    @Test
    public void fromCallableTest() throws InterruptedException {
        Flowable.fromCallable(() -> "ssss").subscribe(i -> System.out.println("Received: " + i), e -> e.printStackTrace());

        Flowable.unsafeCreate(emitter -> {
            emitter.onNext("2");
            emitter.onNext("3");
            emitter.onComplete();

        }).subscribe();
    }




}
